package com.gukun.springboot.tdengine.service.datasubscription;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.gukun.springboot.tdengine.dao.TDEngineDao;
import com.gukun.springboot.tdengine.domain.managetag.DbsManagetagInfo;
import com.gukun.springboot.tdengine.dto.DataSubscriptionDto;
import com.gukun.springboot.tdengine.dto.DsDto;
import com.gukun.springboot.tdengine.dto.ShowTableDto;
import com.gukun.springboot.tdengine.dto.TDengineConfigDto;
import com.gukun.springboot.tdengine.pojo.DBConfig;
import com.gukun.springboot.tdengine.service.TDEngineService;
import com.gukun.springboot.tdengine.service.managerpoints.ManagerPointsInfoService;
import com.gukun.springboot.tdengine.service.topic.TdengineTopic;
import com.gukun.springboot.tdengine.service.writeplant.FastWriteTagData;
import com.gukun.springboot.tdengine.service.writeplant.TDConfigDto;
import com.taosdata.jdbc.TSDBDriver;
import com.taosdata.jdbc.tmq.ConsumerRecord;
import com.taosdata.jdbc.tmq.ConsumerRecords;
import com.taosdata.jdbc.tmq.TaosConsumer;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.sql.*;
import java.time.Duration;
import java.util.*;

@Service
public class DataSubscriptionService {

    private FastWriteTagData finalWriteTagData;


//    private List<TaosConsumer<TdengineTopic.ResultTagData>> consumerList;

    private String uuidName = "data-subscription-" + UUID.randomUUID();
    private static final Logger log = LoggerFactory.getLogger(DataSubscriptionService.class);

    private static int RES_COUNT = 0;

    @Autowired
    private TDEngineService tdEngineService;

    /**
     * 查询数据库下的所有主题
     *
     * @param showTableDto
     * @return
     */
    public Page<DataSubscriptionDto> showTopics(ShowTableDto showTableDto) throws SQLException {
        DBConfig dbConfig = showTableDto.getDbConfig();
        if (dbConfig != null && StringUtils.isNotBlank(dbConfig.getDbName()) && StringUtils.isNotBlank(dbConfig.getHost())) {
            //从mysql数据库里面查询
            List<DataSubscriptionDto> result = getAllTopicInfo(showTableDto);
            Integer currentPage = showTableDto.getCurrentPage();
            Integer pageSize = showTableDto.getPageSize();
            Page<DataSubscriptionDto> page = new Page<>(currentPage, pageSize);
            page.setTotal(result.size());
            if (currentPage * pageSize > result.size()) {
                result = result.subList((currentPage - 1) * pageSize, result.size());
            } else {
                result = result.subList((currentPage - 1) * pageSize, currentPage * pageSize);
            }
            page.setRecords(result);
            return page;
        }
        return null;
    }

    private List<DataSubscriptionDto> getAllTopicInfo(ShowTableDto showTableDto) throws SQLException {
        List<DataSubscriptionDto> result = new ArrayList<>();
        DBConfig dbConfig = showTableDto.getDbConfig();
        TDEngineDao tdEngineDao = tdEngineService.getTDEngineDao(dbConfig.getHost(), dbConfig.getPort(), dbConfig.getUsername(), dbConfig.getPassword(), dbConfig.getDbName());
        tdEngineDao.useDb(dbConfig.getDbName());
        String sql = "select * from information_schema.ins_topics where 1 = 1 ";
        if (StringUtils.isNotBlank(dbConfig.getDbName())) {
            sql += " and db_name =  '" + dbConfig.getDbName() + "' ";
        }

        if (StringUtils.isNotBlank(showTableDto.getTableName())) {
            sql += " and topic_name like  '%" + showTableDto.getTableName() + "%' ";
        }

        List<List<String>> list = tdEngineDao.executeSQL(sql, 0);
        if (list != null && list.size() > 0) {
            Map<Integer, String> columnMap = new HashMap<>();
            JSONObject jsonObject = null;
            for (int i = 0; i < list.size(); i++) {
                if (i == 0) {
                    List<String> columnList = list.get(0);
                    int index = 0;
                    for (String columnName : columnList) {
                        columnMap.put(index, columnName);
                        index++;
                    }
                } else {
                    jsonObject = new JSONObject();
                    List<String> columnList = list.get(i);
                    int index = 0;
                    for (String val : columnList) {
                        String columnName = columnMap.get(index);
                        jsonObject.put(columnName, val);
                        index++;
                    }
                    DataSubscriptionDto myObject = new DataSubscriptionDto();
                    myObject.setCreateTime(jsonObject.getDate("create_time"));
                    myObject.setSql(jsonObject.getString("sql"));
                    myObject.setTopicName(jsonObject.getString("topic_name"));
                    myObject.setDbName(jsonObject.getString("db_name"));
                    result.add(myObject);
                }
            }
        }
        return result;
    }

    /**
     * 删除主题
     *
     * @return
     */
    public String deleteTopics(DsDto dsDto) throws SQLException {
        if (dsDto != null && dsDto.getDataList() != null && dsDto.getDbConfig() != null) {
            List<String> topicNameList = new ArrayList<>();
            DBConfig dbConfig = dsDto.getDbConfig();
            TDEngineDao tdEngineDao = tdEngineService.getTDEngineDao(dbConfig.getHost(), dbConfig.getPort(), dbConfig.getUsername(), dbConfig.getPassword(), dbConfig.getDbName());
            tdEngineDao.useDb(dbConfig.getDbName());
            for (DataSubscriptionDto dataSubscriptionDto : dsDto.getDataList()) {
                String sql = " drop topic  " + dataSubscriptionDto.getTopicName();
                tdEngineDao.executeUpdateSQL(sql);
            }
            return "删除成功";
        }
        return "";
    }

    /**
     * 数据订阅-数据同步
     *
     * @param dsDto
     * @return
     */
    public String syncDataSubInfo(DsDto dsDto) throws Exception {
        if (dsDto != null && dsDto.getDataList() != null && dsDto.getDbConfig() != null && dsDto.gettDengineConfigDto() != null) {
            Connection connection = initConnection(dsDto.getDbConfig());
            Statement statement = connection.createStatement();
            statement.executeUpdate("USE " + dsDto.getDbConfig().getDbName());

            //订阅的主题列表
            Set<String> topics = new HashSet<>();
            List<DataSubscriptionDto> dataList = dsDto.getDataList();
            for (DataSubscriptionDto dataSubscriptionDto : dataList) {
                String topicName = dataSubscriptionDto.getTopicName();
                topics.add(topicName);
            }

            for (String topicName : topics) {
                List<TaosConsumer<TdengineTopic.ResultTagData>> consumerList = new ArrayList<>();
                consumerList = getConsumer(dsDto.getDbConfig(), topicName);
                if (consumerList != null) {
                    pollExample(consumerList, dsDto);
                }
                System.out.println("pollExample executed successfully.");
            }


            return "数据订阅-数据同步";
        }
        return "";
    }

    private Thread thread;

    private void pollExample(List<TaosConsumer<TdengineTopic.ResultTagData>> consumerList, DsDto dsDto) throws Exception {
        if (thread != null && thread.isAlive()) {
            log.warn("线程已经存在，不退出。");
            return;
        }
        List<DataSubscriptionDto> dataList = dsDto.getDataList();
        if (dataList != null && dataList.size() > 0) {
            List<String> topics = new ArrayList<>();
            for (DataSubscriptionDto dataSubscriptionDto : dataList) {
                String topicName = dataSubscriptionDto.getTopicName();
                topics.add(topicName);
            }
            if (topics != null && topics.size() > 0) {
                for (TaosConsumer<TdengineTopic.ResultTagData> consumer : consumerList) {
                    consumer.subscribe(topics);
                    System.out.println("Subscribe topics successfully." + String.join(",", topics));
                    //多线程进行消费
                    thread = new Thread(new Runnable() {
                        @Override
                        public void run() {
                            String time = new DateTime().toString();
                            int count = 0;
                            String mockData = "";
                            TdengineTopic.ResultTagData value = null;
                            Timestamp ts = null;
                            String tbname = null;
                            String v = null;

//                            TDConfigDto tdConfigDto = new TDConfigDto();
//                            tdConfigDto.setDbName(dsDto.gettDengineConfigDto().getName());
//                            tdConfigDto.setIp(dsDto.gettDengineConfigDto().getHost());
//                            tdConfigDto.setPort(dsDto.gettDengineConfigDto().getPort());
//                            tdConfigDto.setUsername(dsDto.gettDengineConfigDto().getUsername());
//                            tdConfigDto.setPassword(dsDto.gettDengineConfigDto().getPassword());
//                            tdConfigDto.setTablePre("_");
//                finalWriteTagData =  new FastWriteTagData(tdConfigDto,false,uuidName);

                            while (true) {
                                System.out.println("Polling for records at " + DateTime.now());
                                long d1 = System.currentTimeMillis();
                                ConsumerRecords<TdengineTopic.ResultTagData> records = null;
                                try {
                                    records = consumer.poll(Duration.ofMillis(1000));
                                } catch (SQLException e) {
                                    throw new RuntimeException("消费拉取异常");
                                }
                                long d2 = System.currentTimeMillis();
                                count += records.count();
                                //
                                RES_COUNT = count;

                                System.out.println("Pulled " + count + " records.");
                                System.out.println("Poll duration: " + (d2 - d1) + " ms.");
                                int start = 0;
                                for (ConsumerRecord<TdengineTopic.ResultTagData> data : records) {
                                    start++;
                                    value = data.value();
                                    ts = value.getTs();
                                    tbname = value.getTbname();
                                    v = value.getV().toString();
                                    mockData = tbname + "," + ts.getTime() + "," + v + ",0";
                                    System.out.println(tbname + "," + ts + "," + v);
                                    if (start > 20) {
                                        break;
                                    }
                                }

//                                int index = 1;
////                    int allSize = finalWriteTagData.getTaskQueues(uuidName).size();
//                                for (ConsumerRecord<TdengineTopic.ResultTagData> record : records) {
//                                    value = record.value();
//                                    ts = value.getTs();
//                                    tbname = value.getTbname();
//                                    v = value.getV().toString();
//                                    mockData = tbname + "," + ts.getTime() + "," + v + ",0";
//                                    if (tbname.equalsIgnoreCase("tag_1")) {
//                                        System.out.println(tbname + "," + ts + "," + v);
//                                    }
////                        int queueId = index%allSize;
////                        finalWriteTagData.getTaskQueues(uuidName).get(queueId).put(mockData);
//                                    index++;
//                                }
                            }
                        }
                    });
                    thread.start();
                }
            }
        }
    }


    /**
     * 关闭订阅数据同步
     *
     * @param dsDto
     * @return
     */
    public String closeSyncDataSubInfo(DsDto dsDto) throws Exception {
        thread.interrupt();
//        if (dsDto != null && dsDto.getDataList() != null && dsDto.getDbConfig() != null && dsDto.gettDengineConfigDto() != null) {
//            Connection connection = initConnection(dsDto.getDbConfig());
//            Statement statement = connection.createStatement();
//            statement.executeUpdate("USE " + dsDto.getDbConfig().getDbName());
//
//            //订阅的主题列表
//            Set<String> topics = new HashSet<>();
//            List<DataSubscriptionDto> dataList = dsDto.getDataList();
//            for (DataSubscriptionDto dataSubscriptionDto : dataList) {
//                String topicName = dataSubscriptionDto.getTopicName();
//                topics.add(topicName);
//            }
//            List<TaosConsumer<TdengineTopic.ResultTagData>> consumerList = null;
//            if (topics != null && topics.size() > 0) {
//                consumerList = getConsumer(dsDto.getDbConfig(),String.join(",",topics));
//            }
//            if (consumerList != null) {
//                for (TaosConsumer<TdengineTopic.ResultTagData> consumer : consumerList) {
//                    consumer.unsubscribe();
//                }
//            }
        System.out.println("pollExample unsubscribe successfully.");
//            Thread.sleep(1000);
//        }
//        if (finalWriteTagData != null) {
//            finalWriteTagData.stopAll(uuidName);
//            finalWriteTagData = null;
//        }
        return "关闭订阅结束";
    }

    private List<TaosConsumer<TdengineTopic.ResultTagData>> getConsumer(DBConfig dbConfig, String topicName) {
        Properties config = new Properties();
        config.setProperty("td.connect.type", "ws");
        config.setProperty("bootstrap.servers", dbConfig.getHost() + ":" + dbConfig.getPort());
        config.setProperty("auto.offset.reset", "latest");
        config.setProperty("msg.with.table.name", "true");
        config.setProperty("enable.auto.commit", "true");
        config.setProperty("auto.commit.interval.ms", "500");
        config.setProperty("group.id", uuidName);
        config.setProperty("client.id", uuidName);
        config.setProperty("td.connect.user", dbConfig.getUsername());
        config.setProperty("td.connect.pass", dbConfig.getPassword());
        config.setProperty("value.deserializer", "com.gukun.springboot.tdengine.service.topic.TdengineTopic$ResultDeserializer");
        config.setProperty("value.deserializer.encoding", "UTF-8");

        try {
            List<TaosConsumer<TdengineTopic.ResultTagData>> list = new ArrayList<>();
//            for(int i = 0;i<32;i++){
            config.setProperty("client.id", uuidName);
            TaosConsumer<TdengineTopic.ResultTagData> consumer = new TaosConsumer<>(config);
            list.add(consumer);
//            }

            return list;
        } catch (Exception ex) {

        }
        return null;
    }

    private Connection initConnection(DBConfig dbConfig) {
        String url = "jdbc:TAOS-WS://" + dbConfig.getHost() + ":" + dbConfig.getPort() + "?user=" + dbConfig.getUsername() + "&password=" + dbConfig.getPassword();
        Properties properties = new Properties();
        properties.setProperty(TSDBDriver.PROPERTY_KEY_LOCALE, "C");
        properties.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8");

        try {
            Connection connection = DriverManager.getConnection(url, properties);
            return connection;
        } catch (SQLException ex) {
            System.out.println("Failed to create connection, url:" + url + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage());
        }
        return null;
    }

    /**
     * 同步历史数据
     *
     * @param dsDto
     * @return
     */
    public String syncDataInfo(DsDto dsDto) {
        if (dsDto != null && dsDto.getDataList() != null) {
            DataSubscriptionDto dto = dsDto.getDataList().get(0);
            String tableName = dto.getTable_name();
            DateTime startDatetime = dsDto.gettDengineConfigDto().getStartDatetime();
            DateTime endDatetime = dsDto.gettDengineConfigDto().getEndDatetime();

            TDengineConfigDto tDengineConfigDto = dsDto.gettDengineConfigDto();
            String targetTableName = tDengineConfigDto.getTargetTableName();
            try {
                if (startDatetime == null || endDatetime == null || startDatetime.isAfter(endDatetime)) {
                    //同步所有数据
                    try (Connection connection = initConnection(dsDto.getDbConfig()); Statement statement = connection.createStatement()) {
                        statement.executeUpdate("USE " + dsDto.getDbConfig().getDbName());
                        String sql = "select ts,v from " + dsDto.getDbConfig().getDbName() + "." + tableName + " order by ts asc";
                        getResult(tDengineConfigDto, targetTableName, statement, sql);
                    }
                } else {
                    //同步查询结果数据
                    try (Connection connection = initConnection(dsDto.getDbConfig()); Statement statement = connection.createStatement()) {
                        statement.executeUpdate("USE " + dsDto.getDbConfig().getDbName());
                        String sql = "select ts,v from " + dsDto.getDbConfig().getDbName() + "." + tableName + " where ts>='" + startDatetime + "' and ts <='" + endDatetime + "' order by ts desc";
                        getResult(tDengineConfigDto, targetTableName, statement, sql);
                    }
                }
            } catch (Exception ex) {
                log.error("Error during data synchronization: {}", ex.getMessage(), ex);
            }
        }

        return "success";
    }

    private void getResult(TDengineConfigDto tDengineConfigDto, String targetTableName, Statement statement, String
            sql) throws SQLException {
        try (ResultSet resultSet = statement.executeQuery(sql)) {
            String insertSql = "insert into " + targetTableName + " values";
            int size = 0;
            while (resultSet.next()) {
                Long ts = resultSet.getLong(1);
                String val = resultSet.getString(2);
                insertSql += "(" + ts + "," + val + ") ";
                size++;
                if (size == 4000) {
                    doInsert(tDengineConfigDto, insertSql);
                    insertSql = "insert into " + targetTableName + " values";
                    size = 0;
                }
            }
            if (size > 0) {
                doInsert(tDengineConfigDto, insertSql);
            }
        } catch (Exception ex) {
            log.error("Error executing SQL: {}", sql, ex);
            throw new SQLException("Error executing SQL: " + sql, ex);
        }

    }

    private void doInsert(TDengineConfigDto dbConfig, String insertSql) throws SQLException {
        TDEngineDao tdEngineDao = tdEngineService.getTDEngineDao(dbConfig.getHost(), dbConfig.getPort(), dbConfig.getUsername(), dbConfig.getPassword(), dbConfig.getName());
        tdEngineDao.useDb(dbConfig.getName());

        tdEngineDao.executeUpdateSQL(insertSql);
    }

    public String getResCount() {
        return String.valueOf(RES_COUNT);
    }
}
